#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include <errno.h>
#include <zlib.h>
#include <vector>
#include <iostream>

#include "DataSvr.h"
#include "DataSvrClient.h"
#include "DataSvrCommon.h"
#include "DataSvrHelper.h"
#include "DataSvrProtoHelper.h"
#include "ProtoHelper.h"
#include "SocketHelper.h"
#include "HashHelper.h"
#include "mycrc32.h"

using std::vector;
using std::cout;
using std::endl;

LoggerPtr DataSvr::_logger = Logger::getLogger("DataSvr");

#define NOTIFY_MDS_FILE_INFO_REASON_FILE_OP 0
#define NOTIFY_MDS_FILE_INFO_REASON_REPLICATE 1

int DataSvr::init(const char * log_conf_file, int my_ds_id, const char * cs_ip, unsigned short cs_port, \
                                int cs_conn_pool_size, int cs_socket_conn_timeout, int cs_socket_send_timeout, \
                                int cs_socket_recv_timeout, int cs_socket_max_response_size) {
	if (NULL == cs_ip ||  NULL == log_conf_file || my_ds_id < 0) {
                LOG4CXX_ERROR(_logger, "init error, invalid args");
                return -1;
        }
	srand(time(NULL));
	_sfw.init_logger(log_conf_file);
        int ret = _csc.init(cs_ip, cs_port, cs_conn_pool_size, cs_socket_conn_timeout, cs_socket_send_timeout, \
                                        cs_socket_recv_timeout, cs_socket_max_response_size);
        if (ret < 0) {
                LOG4CXX_ERROR(_logger, __FUNCTION__ << ", cnf svr init error, ret:" << ret);
                return -2;
        }
	list<DS_CNF> l_ds;
	ret = _csc.get_all_ds_cnf(l_ds);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not get all ds cnf, ret:" << ret);
		return -3;
	}
	bool found = false;
	for(list<DS_CNF>::const_iterator iter = l_ds.begin(); l_ds.end() != iter; iter++) {
		if (iter->id == my_ds_id) {
			_listen_ip = iter->ip;
			_listen_port = iter->port;
			_thread_num = iter->thread_num;
			_log_conf_file = log_conf_file;
			_socket_idle_timeout = iter->socket_idle_timeout;
			_socket_send_timeout = iter->socket_send_timeout;
			_socket_recv_timeout = iter->socket_recv_timeout;
			_max_request_size = iter->socket_max_request_size;
			_root_data_path = iter->root_data_path;
			_my_ds_id = my_ds_id;
			_my_mds_id = iter->mds_id;
			ret = _sfw.init(iter->thread_num, handle_socket_input_func, this, iter->ip.c_str(), iter->port, \
					iter->socket_idle_timeout, NULL, NULL);
			found = true;
			if (ret < 0) {
				LOG4CXX_ERROR(_logger, "sfw init error, ret:" << ret);
				return -4;
			}

		}
	}
	if (!found) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not get my id from conf server");
		return -5;
	}
	
	ret = _dscg.init(&_csc);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", data svr client group init error, ret:" << ret);
		return -6;
	}
	ret = _mdscg.init(&_csc);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", meta data svr client group init error, ret:" << ret);
		return -7;
	}
	pthread_mutex_init(&_replicate_tasks_mutex, NULL);
	pthread_cond_init(&_replicate_tasks_cond,  NULL);
	pthread_create(&_replicate_thread, NULL, replicate_thread_routine, this);
        LOG4CXX_INFO(_logger, "init DataSvr OK, listen_ip:" << _listen_ip << ", listen_port:" << _listen_port << \
				", thread_num:" << \
                                _thread_num << ", log_conf_file:" << _log_conf_file << ", socket_idle_timeout:" << \
                                _socket_idle_timeout << ", socket_send_timeout:" << _socket_send_timeout << \
                                ", socket_recv_timeout:" << _socket_recv_timeout << ", max_request_size:" << _max_request_size << \
                                ", root_data_path:" << _root_data_path);
        return 1;
}

int DataSvr::put_replicate_task_to_queue(REPLICATE_TASK task) {
	if (task.dir_id < 0 || task.file_id < 0 || task.src_ds_id < 0 || task.src_file_size < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", invalid args");
		return -1;
	}
	pthread_mutex_lock(&_replicate_tasks_mutex);
	if (_replicate_tasks.find(task) == _replicate_tasks.end()) {
		_replicate_tasks.insert(task);
		pthread_cond_signal(&_replicate_tasks_cond);
		pthread_mutex_unlock(&_replicate_tasks_mutex);
		return 1;
	}
	pthread_mutex_unlock(&_replicate_tasks_mutex);
	return -2;
}

int DataSvr::get_replicate_task_from_queue(REPLICATE_TASK & task) {
	pthread_mutex_lock(&_replicate_tasks_mutex);
	pthread_cond_wait(&_replicate_tasks_cond, &_replicate_tasks_mutex);
	if (_replicate_tasks.size() < 1) {
		pthread_mutex_unlock(&_replicate_tasks_mutex);
		return -1;
	}
	set<REPLICATE_TASK>::const_iterator iter = _replicate_tasks.begin();
	task = *iter;
	_replicate_tasks.erase(task);
	pthread_mutex_unlock(&_replicate_tasks_mutex);
	return 1;
}

void * DataSvr::replicate_thread_routine(void * arg) {
	DataSvr * ds = (DataSvr *)arg;
	while (true) {
		REPLICATE_TASK t;
		int ret = ds->get_replicate_task_from_queue(t);
		if (ret < 0) continue;
		ret = ds->replicate(t.dir_id, t.file_id, t.src_ds_id, t.src_file_size, t.src_crcs);
		if (ret < 0) {
			LOG4CXX_ERROR(_logger, __FUNCTION__ << ", replicate error, dir_id:" << t.dir_id << ", file_id:" << \
					t.file_id << ", src_ds_id:" << t.src_ds_id << ", src_file_size:" << t.src_file_size);
		}
		else {
			LOG4CXX_INFO(_logger, __FUNCTION__ << ", ok, dir_id:" << t.dir_id << ", file_id:" << \
					t.file_id << ", src_ds_id:" << t.src_ds_id << ", src_file_size:" << t.src_file_size);
		}
	}
}

int DataSvr::start() {
	return _sfw.start_server();
}

int DataSvr::handle_socket_input_func(int fd, void * arg) {
        DataSvr * ds = (DataSvr *)arg;
        json_object * j;
        int cmd_id;
        string request;
        int ret = recv_and_parse_request(fd, ds->_socket_recv_timeout, ds->_max_request_size, j, cmd_id, request);
        if (ret < 0) {
                LOG4CXX_ERROR(_logger, "recv and parse request error, ret:" << ret);
                return -1;
        }
        string response, extra_response;
	bool is_network_error = false;
        ds->dispatch_request(cmd_id, j, response, extra_response, fd, is_network_error);
        json_object_put(j);
	if (is_network_error) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", detected network error while processing request:" << request << \
				", response:" << response);
		return -2;
	}
        if (response.size() < 1) response = build_error_response(DS_ERR_UNKNOWN_ERR);
        ret = send_cmd(fd, ds->_socket_send_timeout, response.data(), response.size());
        if (ret < 0) {
                LOG4CXX_ERROR(_logger, "reply cmd error, ret:" << ret << ", fd:" << fd << ", request:" << request << \
                                        ", response:" << response);
                return -3;
        }
	if (extra_response.size() > 0) {
		ret = send_all(fd, extra_response.c_str(), extra_response.size(), ds->_socket_send_timeout);
		if (ret < 0) {
			LOG4CXX_ERROR(_logger, "reply extra reponse error, ret:" << ret << ", fd:" << fd << ", request:" << \
						request << ", extra response size:" << extra_response.size());
			return -4;
		}
	}
	LOG4CXX_DEBUG(_logger, __FUNCTION__ << ", ok, cmd_id:" << cmd_id);
        return 2;
}

int DataSvr::dispatch_request(int cmd_id, json_object * j, string & response, string & extra_response, const int fd, \
			bool & is_network_error) {
        int ret = -1;
        switch (cmd_id) {
                case DS_CMD_CREATE_FILE:
                        ret = handle_create_file(j, response, extra_response, fd, is_network_error);
                        break;
                case DS_CMD_DELETE_FILE:
                        ret = handle_delete_file(j, response, extra_response, fd, is_network_error);
                        break;
                case DS_CMD_READ_FILE:
                        ret = handle_read_file(j, response, extra_response, fd, is_network_error);
                        break;
                case DS_CMD_WRITE_FILE:
                        ret = handle_write_file(j, response, extra_response, fd, is_network_error);
                        break;
		case DS_CMD_GET_FILE_CONTROL_INFO:
			ret = handle_get_file_control_info(j, response, extra_response, fd, is_network_error);
			break;
		case DS_CMD_REPLICATE:
			ret = handle_replicate(j, response, extra_response, fd, is_network_error);
			break;
        	case DS_CMD_TRUNCATE:
			ret = handle_truncate(j, response, extra_response, fd, is_network_error);
			break;
                default:
                        LOG4CXX_ERROR(_logger, "unknown command, cmd_id:" << cmd_id);
                        ret = -100;
        }
        return ret;
}

int DataSvr::handle_get_file_control_info(json_object * j, string & response, string & extra_response, \
			const int fd, bool & is_newwork_error) {
        json_object * j_dir_id = json_object_object_get(j, "DIR_ID");
        json_object * j_file_id = json_object_object_get(j, "FILE_ID");
        if ( is_error(j_dir_id) || is_error(j_file_id) ) {
                LOG4CXX_ERROR(_logger, __FUNCTION__ << ", arg missing:" << json_object_to_json_string(j));
                response = build_error_response(DS_ERR_INVALID_ARGS, "");
                return -1;
        }
	int dir_id = json_object_get_int(j_dir_id);
	int64_t file_id = json_object_get_int(j_file_id);
	string content_file_path = get_file_hash_path(_root_data_path.c_str(), dir_id, file_id);
	string control_file_path = get_control_file_hash_path(_root_data_path.c_str(), dir_id, file_id);
	vector<uint32_t> crcs;
	int ret = parse_control_file(control_file_path.c_str(), crcs);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", parse control file error, file:" << control_file_path << ", ret:" << \
				ret);
                response = build_error_response(DS_ERR_OS_FILE_OP_ERR);
		return -2;
	}
	struct stat s;
	ret = stat(content_file_path.c_str(), &s);
	if (0 != ret) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", stat file error, file:" << content_file_path << ", errno:" << errno);
		return -3;
	}
	response = build_get_file_control_info_response(crcs, s.st_size);
	LOG4CXX_INFO(_logger, __FUNCTION__ << ", ok, file:" << control_file_path << ", response:" << response);
	return 1;
}
	
int DataSvr::handle_replicate(json_object * j, string & response, string & extra_response, \
			const int fd, bool & is_newwork_error) {
        json_object * j_dir_id = json_object_object_get(j, "DIR_ID");
        json_object * j_file_id = json_object_object_get(j, "FILE_ID");
        json_object * j_src_ds_id = json_object_object_get(j, "SRC_DS_ID");
        json_object * j_src_file_size = json_object_object_get(j, "SRC_FILE_SIZE");
        json_object * j_src_crc = json_object_object_get(j, "SRC_CRC");
        if ( NULL == j_dir_id || NULL == j_file_id || NULL == j_src_ds_id || NULL == j_src_file_size || NULL == j_src_crc ) {
                LOG4CXX_ERROR(_logger, __FUNCTION__ << ", arg missing:" << json_object_to_json_string(j));
                response = build_error_response(DS_ERR_INVALID_ARGS, "");
                return -1;
        }
	int32_t dir_id = json_object_get_int(j_dir_id);
	int64_t file_id = json_object_get_int(j_file_id);
	int32_t src_ds_id = json_object_get_int(j_src_ds_id);
	int64_t src_file_size = json_object_get_int(j_src_file_size);
	vector<uint32_t> src_crcs;
	for(int i=0; i < json_object_array_length(j_src_crc); i++) {
                json_object * j_t = json_object_array_get_idx(j_src_crc, i);
                src_crcs.push_back(json_object_get_int(j_t));
        }
	REPLICATE_TASK t;
	t.dir_id = dir_id;
	t.file_id = file_id;
	t.src_ds_id = src_ds_id;
	t.src_file_size = src_file_size;
	t.src_crcs = src_crcs;
	put_replicate_task_to_queue(t);
	response = build_ok_response();
	LOG4CXX_INFO(_logger, __FUNCTION__ << ", ok, request:" << json_object_to_json_string(j));
	return 1;
}

int DataSvr::handle_create_file(json_object * j, string & response, string & extra_response, \
			const int fd, bool & is_newwork_error) {
        json_object * j_dir_id = json_object_object_get(j, "DIR_ID");
        json_object * j_file_id = json_object_object_get(j, "FILE_ID");
        if ( is_error(j_dir_id) || is_error(j_file_id) ) {
                LOG4CXX_ERROR(_logger, "in handle create file, args missing:" << json_object_to_json_string(j));
                response = build_error_response(DS_ERR_INVALID_ARGS, "");
                return -1;
        }
	int dir_id = json_object_get_int(j_dir_id);
	int64_t file_id = json_object_get_int(j_file_id);
	int ret = create_new_file(dir_id, file_id); 
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__  << ", create new file error, requerst:" << json_object_to_json_string(j) << \
					", ret:" << ret);
                response = build_error_response(DS_ERR_OS_FILE_OP_ERR, "");
		return -2;
	}
	ret = notify_mds_file_info(dir_id, file_id, 0, NOTIFY_MDS_FILE_INFO_REASON_FILE_OP);
        if (ret < 0) {
                LOG4CXX_ERROR(_logger, __FUNCTION__ << ", notify_mds_file_info error, dir_id:" << dir_id << \
                                ", file_id:" << file_id);
		response = build_error_response(DS_ERR_NOTIFY_MDS_ERROR);
                return -3;
        }

	response = build_ok_response();
	LOG4CXX_INFO(_logger, __FUNCTION__ << ", ok, request:" << json_object_to_json_string(j));
	return 1;
}

int DataSvr::create_new_file(int32_t dir_id, int64_t file_id) {
	if (dir_id < 0 || file_id < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", invalid args");
		return -1;
	}
	string content_file_path = get_file_hash_path(_root_data_path.c_str(), dir_id, file_id);
	string control_file_path = get_control_file_hash_path(_root_data_path.c_str(), dir_id, file_id);
	struct stat s;
	int ret = stat(content_file_path.c_str(), &s);
	if (0 == ret) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", file already exist, file:" << content_file_path << ", errno:" << errno);
		return -2;
	}
	else if (ENOENT != errno) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", stat error, file:" << content_file_path << ", errno:" << errno);
		return -3;
	}
	ret = stat(control_file_path.c_str(), &s);
	if (0 == ret) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", file already exist, file:" << control_file_path);
		return -4;
	}
	else if (ENOENT != errno) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", stat error, file:" << control_file_path << ", errno:" << errno);
		return -5;
	}
	FILE * fp1 = fopen(content_file_path.c_str(), "w");
	if (NULL == fp1) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", fopen error, file:" << content_file_path << ", errno:" << errno);
		return -6;
	}
	fclose(fp1);

	ret = write_control_file(control_file_path.c_str(), vector<uint32_t>()); 
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__  << ", fwrite control file error, file:" << control_file_path << ", ret:" << \
				ret); 
		return -7;
	}
	LOG4CXX_INFO(_logger, __FUNCTION__ << ", ok, path:" << content_file_path);
	return 1;
}


int DataSvr::handle_delete_file(json_object * j, string & response, string & extra_response, \
				const int fd, bool & is_newwork_error) {
        json_object * j_dir_id = json_object_object_get(j, "DIR_ID");
        json_object * j_file_id = json_object_object_get(j, "FILE_ID");
        if ( is_error(j_dir_id) || is_error(j_file_id) ) {
                LOG4CXX_ERROR(_logger, __FUNCTION__ << ", args missing:" << json_object_to_json_string(j));
                response = build_error_response(DS_ERR_INVALID_ARGS, "");
                return -1;
        }
	int dir_id = json_object_get_int(j_dir_id);
	int64_t file_id = json_object_get_int(j_file_id);
	string content_file_path = get_file_hash_path(_root_data_path.c_str(), dir_id, file_id);
	string control_file_path = get_control_file_hash_path(_root_data_path.c_str(), dir_id, file_id);
	int ret = unlink(content_file_path.c_str());
	if (0 != ret) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", unlink error, file:" << content_file_path);
		response = build_error_response(DS_ERR_OS_FILE_OP_ERR);
		return -2;
	}
	ret = unlink(control_file_path.c_str());
	if (0 != ret) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", unlink error, file:" << control_file_path);
		response = build_error_response(DS_ERR_OS_FILE_OP_ERR);
		return -3;
	}
	ret = notify_mds_file_info(dir_id, file_id, 1, NOTIFY_MDS_FILE_INFO_REASON_FILE_OP);
        if (ret < 0) {
                LOG4CXX_ERROR(_logger, __FUNCTION__ << ", notify_mds_file_info error, dir_id:" << dir_id << \
                                ", file_id:" << file_id);
		response = build_error_response(DS_ERR_NOTIFY_MDS_ERROR);
                return -4;
        }
	response = build_ok_response();
	LOG4CXX_INFO(_logger, __FUNCTION__ << ", ok, file:" << content_file_path);
	return 1;
}

int DataSvr::handle_read_file(json_object * j, string & response, string & extra_response, \
				const int fd, bool & is_newwork_error) {
        json_object * j_dir_id = json_object_object_get(j, "DIR_ID");
        json_object * j_file_id = json_object_object_get(j, "FILE_ID");
        json_object * j_begin_pos = json_object_object_get(j, "BEGIN_POS");
        json_object * j_len = json_object_object_get(j, "LEN");
        if( is_error(j_dir_id) || is_error(j_file_id) || is_error(j_begin_pos) || is_error(j_len)) {
                LOG4CXX_ERROR(_logger, __FUNCTION__ << ", args missing:" << json_object_to_json_string(j));
                response = build_error_response(DS_ERR_INVALID_ARGS, "");
                return -1;
        }
	int dir_id = json_object_get_int(j_dir_id);
	int64_t file_id = json_object_get_int(j_file_id);
	int64_t begin_pos = json_object_get_int(j_begin_pos);
	int len = json_object_get_int(j_len);
	string content_file_path = get_file_hash_path(_root_data_path.c_str(), dir_id, file_id);
	string control_file_path = get_control_file_hash_path(_root_data_path.c_str(), dir_id, file_id);
	if (0 == len) {
		response = build_ok_response();
		LOG4CXX_INFO(_logger, __FUNCTION__ << ", read file:" << content_file_path << ", len:0");
		return 1;
	}

	FILE * fp = fopen(content_file_path.c_str(), "r");
	if (NULL == fp) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", foepn error, file:" << content_file_path.c_str() << ", errno:" << errno);
                response = build_error_response(DS_ERR_OS_FILE_OP_ERR);
		return -2;
	}
	char * buf = new char[len];
	int ret = fseek(fp, begin_pos, SEEK_SET);
	if (0 != ret) {
		fclose(fp);
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", fseek error, file:" << content_file_path << ", begin_pos:" << \
				begin_pos  << ", len:" << len << ", errno:" << errno);
                response = build_error_response(DS_ERR_OS_FILE_OP_ERR);
		return -3;
	}
	ret = fread(buf, len, 1, fp);
	fclose(fp);
	if (1 != ret) {
		delete[] buf;
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", fread error, file:" << content_file_path << ", begin_pos:" << \
				begin_pos << ", len:" << len << ", erron:" << errno);
                response = build_error_response(DS_ERR_OS_FILE_OP_ERR);
		return -4;
	}
	response = build_ok_response();
	extra_response.assign(buf, len);
	//cout << "extra read response:" << extra_response << endl;
	delete[] buf;
	LOG4CXX_INFO(_logger, __FUNCTION__ << ", ok, read file:" << content_file_path << ", begin_pos:" << begin_pos << \
			", len:" << len);
	return 2;
}

int DataSvr::write_file(int32_t dir_id, int64_t file_id, int64_t begin_pos, const char * buf, int64_t len, int32_t mode) {
	if (dir_id < 0 || file_id < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", invalid args");
		return -1;
	}
	string content_file_path = get_file_hash_path(_root_data_path.c_str(), dir_id, file_id);
	string control_file_path = get_control_file_hash_path(_root_data_path.c_str(), dir_id, file_id);
	if (0 == len) {
		return 1;
	}
	int64_t real_begin_pos = begin_pos;
	char * m = "r+";
	struct stat s;
	int ret = stat(content_file_path.c_str(), &s);
	if (DS_WRITE_MODE_APPEND == mode) {
		m = "a";
		if (0 != ret) {
			LOG4CXX_ERROR(_logger, __FUNCTION__ << ", stat error, file:" << content_file_path << ", errno:" << errno);
			return -2;
		}
		real_begin_pos = s.st_size;
	}
	else if (begin_pos > s.st_size) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", there is a write hole in the file:" << content_file_path << \
				", file size:" << s.st_size << ", begin pos:" << begin_pos);
		return -3;
	}
			
	FILE * fp = fopen(content_file_path.c_str(), m);
	if (NULL == fp) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", fopen error, file:" << content_file_path << ", errno:" << errno);
		return -4;
	}
	if (DS_WRITE_MODE_APPEND != mode) {
		int ret = fseek(fp, begin_pos, SEEK_SET);
		if (0 != ret) {
			fclose(fp);
			LOG4CXX_ERROR(_logger, __FUNCTION__ << ", fseek error, file:" << content_file_path << ", begin_pos:" << \
					begin_pos  << ", len:" << len << ", errno:" << errno);
			return -5;
		}
	}
	ret = fwrite(buf, len, 1, fp);
	fclose(fp);
	if (1 != ret) {
			LOG4CXX_ERROR(_logger, __FUNCTION__ << ", fwrite error, file:" << content_file_path << ", begin_pos:" << \
					begin_pos  << ", len:" << len << ", errno:" << errno);
			return -6;

	}
	ret = rebuild_crc(dir_id, file_id, real_begin_pos, len);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", rebuild crc error, file:" << content_file_path << ", begin_pos:" << \
				begin_pos << ", len:" << len << ", ret:" << ret);
		return -7;
	}
	LOG4CXX_INFO(_logger, __FUNCTION__ << ", write ok, file:" << content_file_path << ", len:" << len << ", begin_pos:" << \
				begin_pos << ", mode:" << mode);
	return 2;
}

int DataSvr::handle_write_file(json_object * j, string & response, string & extra_response, const int fd, bool & is_network_error) {
        json_object * j_dir_id = json_object_object_get(j, "DIR_ID");
        json_object * j_file_id = json_object_object_get(j, "FILE_ID");
        json_object * j_begin_pos = json_object_object_get(j, "BEGIN_POS");
        json_object * j_len = json_object_object_get(j, "LEN");
        json_object * j_mode = json_object_object_get(j, "MODE");
        if( is_error(j_dir_id) || is_error(j_file_id) || is_error(j_begin_pos) || is_error(j_len) || is_error(j_mode)) {
                LOG4CXX_ERROR(_logger, __FUNCTION__ << ", args missing:" << json_object_to_json_string(j));
                response = build_error_response(DS_ERR_INVALID_ARGS, "");
                return -1;
        }
	int dir_id = json_object_get_int(j_dir_id);
	int64_t file_id = json_object_get_int(j_file_id);
	int64_t begin_pos = json_object_get_int(j_begin_pos);
	int len = json_object_get_int(j_len);
	int mode  = json_object_get_int(j_mode);
	if (0 == len) {
		response = build_ok_response();
		LOG4CXX_INFO(_logger, __FUNCTION__ << ", write ok, request :" << json_object_to_json_string(j) << ", len:0");
		return 1;
	}
	LOG4CXX_DEBUG(_logger, __FUNCTION__ << ", begin, dir_id:" << dir_id << ", file_id:" << file_id << ", begin_pos:" << \
				begin_pos<< ", len:" << len << ", mode:" << mode);
	char * buf = new char[len];
	int ret = recv_all(fd, buf, len, _socket_recv_timeout);
	if (ret < 1) {
		delete[] buf;
                response = build_error_response(DS_ERR_SERVER_DETECTED_NETWORK_ERR);
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", recv all write content error, ret:" << ret << ", request:" << \
				json_object_to_json_string(j));
		is_network_error = true;
		return -2;
	}
	LOG4CXX_DEBUG(_logger, __FUNCTION__ << ", write data file begin, dir_id:" << dir_id << ", file_id:" << file_id << \
				", begin_pos:" << begin_pos<< ", len:" << len << ", mode:" << mode);
	ret = write_file(dir_id, file_id, begin_pos, buf, len, mode);
	delete[] buf;
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", write file error, dir_id:" << dir_id << ", file_id:" << file_id << \
				", ret:" << ret); 
		response = build_error_response(DS_ERR_OS_FILE_OP_ERR);
		return -3;
	}
	LOG4CXX_DEBUG(_logger, __FUNCTION__ << ", notify mds begin, dir_id:" << dir_id << ", file_id:" << file_id << \
				", begin_pos:" << begin_pos<< ", len:" << len << ", mode:" << mode);
	ret = notify_mds_file_info(dir_id, file_id, 0, NOTIFY_MDS_FILE_INFO_REASON_FILE_OP);
        if (ret < 0) {
                LOG4CXX_ERROR(_logger, __FUNCTION__ << ", notify_mds_file_info error, dir_id:" << dir_id << \
                                ", file_id:" << file_id);
		response = build_error_response(DS_ERR_NOTIFY_MDS_ERROR);
                return -4;
        }
	response = build_ok_response();
	LOG4CXX_INFO(_logger, __FUNCTION__ << ", write ok, request:" << json_object_to_json_string(j) << ", len:" << \
			len << ", begin_pos:" << begin_pos << ", mode:" << mode);
	return 2;
}

int DataSvr::total_rebuild_crc(const int dir_id, const int64_t file_id) {
	if (dir_id < 0 || file_id < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", invalid args");
		return -1;
	}
	string content_file_path = get_file_hash_path(_root_data_path.c_str(), dir_id, file_id);
	string control_file_path = get_control_file_hash_path(_root_data_path.c_str(), dir_id, file_id);
        struct stat s;
        int ret = stat(content_file_path.c_str(), &s);
        if (0 != ret) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", stat error, file:" << content_file_path << ", errno:" << errno);
		return -2;
	}
	vector<uint32_t> crcs;
	if (s.st_size > 0) {
		char * buf = new char[s.st_size];
		FILE * fp = fopen(content_file_path.c_str(), "r");
		int n = fread(buf, s.st_size, 1, fp);
		fclose(fp);
		if (1 != n) {
			LOG4CXX_ERROR(_logger, __FUNCTION__ << ", fread error, file:" << content_file_path << ", errno:" << errno);
			return -3;
		}
		for (int i = 0; i < s.st_size; i += DS_CRC_CHUNK_SIZE) {
			int len = DS_CRC_CHUNK_SIZE;
			int left = s.st_size - i;
			if (left < DS_CRC_CHUNK_SIZE) len = left;
			uint32_t crc_init = crc32(0L, NULL, 0);
			uint32_t crc = crc32(crc_init, (Bytef *)&buf[i], len);
			crcs.push_back(crc);
		}
		delete[] buf;
	}
	ret = write_crc_to_control_file(control_file_path.c_str(), crcs);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", write new crc to control file error, file:" << control_file_path << \
				", ret:" << ret);
		return -4;
	}
	return 1;
}

int DataSvr::rebuild_crc(const int dir_id, const int64_t file_id, const int64_t write_begin_pos, const int write_len) {
	if (dir_id < 0 || file_id < 0 || write_begin_pos < 0 || write_len < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", args error, dir_id:" << dir_id << ", file_id:" << file_id << \
			", write_begin_pos:" << write_begin_pos << ", write_len:" << write_len);
		return -1;
	}
	string content_file_path = get_file_hash_path(_root_data_path.c_str(), dir_id, file_id);
	string control_file_path = get_control_file_hash_path(_root_data_path.c_str(), dir_id, file_id);
	struct stat s;
	int ret = stat(content_file_path.c_str(), &s);
	if (0 != ret) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", fstat error, file:" << content_file_path << ", errno:" << errno);
		return -2;
	}
	int64_t t1 = write_begin_pos / DS_CRC_CHUNK_SIZE;
	int64_t b1 = t1 * DS_CRC_CHUNK_SIZE;
	int64_t t2 = (write_begin_pos + write_len)/ DS_CRC_CHUNK_SIZE;
	int64_t b2 = (t2 + 1) * DS_CRC_CHUNK_SIZE;
	int len = b2 - b1;
	if (b1 + len > s.st_size) len = s.st_size - b1;
	FILE * fp1 = fopen(content_file_path.c_str(), "r");
	if (NULL == fp1) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", fopen err, file:" << content_file_path << ", errno:" << errno);
		return -3;
	}
	ret = fseek(fp1, b1, SEEK_SET);
	if (0 != ret) {
		fclose(fp1);
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", fseek err, file:" << content_file_path << ", errno:" << errno);
		return -4;
	}
	char * buf = new char[len];
	ret = fread(buf, len, 1, fp1);
	fclose(fp1);
	if (1 != ret) {
		delete[] buf;
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", fread err, file:" << content_file_path << ", errno:" << errno << \
				", write_begin_pos:" << write_begin_pos << ", len:" << len);
		return -5;
	}
	vector<uint32_t> updated_crcs;
	for (int i = 0; i < len; i+= DS_CRC_CHUNK_SIZE)  {
		int c = DS_CRC_CHUNK_SIZE;
		int left = len - i;
		if (left < DS_CRC_CHUNK_SIZE) c = left;
		uint32_t x = get_crc32(&buf[i], c);
		updated_crcs.push_back(x);
		//cout << "i:" << i << ", len:" << len << ", c:" << c << ", crc:" << x << endl;
	}
	delete[] buf;
	vector<uint32_t> old_crcs;
	ret = parse_control_file(control_file_path.c_str(), old_crcs);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", parse control file content error, file:" << control_file_path << \
				", ret:" << ret);	
		return -6;
	}
	int updated_crc_begin_index = write_begin_pos / DS_CRC_CHUNK_SIZE;
	int updated_crc_end_index = ((write_begin_pos + write_len - 1) / DS_CRC_CHUNK_SIZE);
	vector<uint32_t> new_crcs;
	int j = 0;
	for (int i = 0; ; i++) {
		if (i > updated_crc_end_index && i >= old_crcs.size()) break;
		if (i < updated_crc_begin_index || i > updated_crc_end_index) {
			//cout << "push old crc:" << old_crcs[i] << endl;
			new_crcs.push_back(old_crcs[i]);
		}
		else {
			new_crcs.push_back(updated_crcs[j]);
			/*
			cout << "push updated crc:" << updated_crcs[j] << ", updated crc size:" << updated_crcs.size() << \
				", j:" << j << ", updated_crc_end_index:" << updated_crc_end_index << \
				", updated_crc_begin_index:" << updated_crc_begin_index << ", write_begin_pos:" << \
				write_begin_pos << ", write_len:" << write_len << endl;
			*/
			j += 1;
		}
	}
	ret = write_crc_to_control_file(control_file_path.c_str(), new_crcs);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", write new crc to control file error, file:" << control_file_path << \
				", ret:" << ret);
		return -7;
	}
	return 1;
}

int DataSvr::replicate(const int32_t dir_id, const int64_t file_id, const int32_t src_ds_id, const int64_t src_file_size, \
                                const vector<uint32_t> src_crcs) {
	if (dir_id < 0 || file_id < 0 || src_ds_id < 0 || src_file_size < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", invalid args");
		return -1;
	}
	if (src_ds_id == _my_ds_id) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", src_ds_id = my_ds_id = " << _my_ds_id);
		return -2;
	}
	string content_file_path = get_file_hash_path(_root_data_path.c_str(), dir_id, file_id);
	string control_file_path = get_control_file_hash_path(_root_data_path.c_str(), dir_id, file_id);
	struct stat s;
	int ret = stat(content_file_path.c_str(), &s);
	if (0 == ret) {
		if (s.st_size > src_file_size) {
			ret = truncate(content_file_path.c_str(), src_file_size);
			if (0 != ret) {
				LOG4CXX_ERROR(_logger, __FUNCTION__ << ", truncate file error, flie:" << content_file_path << \
						", try to truncate to:" << src_file_size << ", errno:" << errno);
			}
			ret = total_rebuild_crc(dir_id, file_id);
			if (ret < 0) {
				LOG4CXX_ERROR(_logger, __FUNCTION__ << ", total_rebuild_crc error, file:" << \
						content_file_path << ", ret:" << ret)
					return -3;
			}
		}
	}
	else {
		if  (ENOENT != errno) { 
			LOG4CXX_ERROR(_logger, __FUNCTION__ << ", stat error, file:" << content_file_path << ", errno:" << errno);
			return -4;
		}
		ret = create_new_file(dir_id, file_id);
		if (ret < 0) {
			LOG4CXX_ERROR(_logger, __FUNCTION__  << ", create new file error, dir_id:" << dir_id << ", file_id:" << \
					file_id << ", ret:" << ret);
			return -5;
		}
	}
	vector<uint32_t> my_crcs;
	ret = parse_control_file(control_file_path.c_str(), my_crcs);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", parse control file error, file:" << control_file_path.c_str() << \
				", ret:" << ret);
		return -6;
	}
	vector<uint32_t>::const_iterator iter_my_crc = my_crcs.begin();
	long long i = 0;
	DataSvrClient * dsc = _dscg.get_data_svr_client(src_ds_id);
	if (NULL == dsc) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not get data svr client, src_ds_id:" << src_ds_id);
		return -7;
	}
	for (vector<uint32_t>::const_iterator iter_src_crc = src_crcs.begin(); src_crcs.end() != iter_src_crc; \
		iter_src_crc++, i += DS_CRC_CHUNK_SIZE) {
		uint32_t src_crc = *iter_src_crc;
		if (iter_my_crc != my_crcs.end()) {
			uint32_t my_crc = *iter_my_crc;
			if (my_crc == src_crc) {
				iter_my_crc++;
				continue;
			}
		}
		int try_to_read_len = DS_CRC_CHUNK_SIZE;
		long long left = src_file_size - i;
		if (left < DS_CRC_CHUNK_SIZE) try_to_read_len = left;
		char * buf = new char[try_to_read_len];
		ret = dsc->read_file(dir_id, file_id, i, try_to_read_len, buf);
		if (ret < 0) {
			LOG4CXX_ERROR(_logger, __FUNCTION__ << ", read file from other data svr error, dir_id:" << dir_id << \
					", file_id:" << file_id << ", ret:" << ret);
			delete[] buf;
			return -8;
		}
		ret = write_file(dir_id, file_id, i, buf, try_to_read_len, DS_WRITE_MODE_NORMAL);
		delete[] buf;
		if (ret < 0) {
			LOG4CXX_ERROR(_logger, __FUNCTION__ << ", write_file error, dir_id:" << dir_id << ", file_id:" << \
					file_id <<", ret:" << ret);
			return -9;
		}
		LOG4CXX_DEBUG(_logger, __FUNCTION__ << ", dir_id:" << dir_id << ", file_id:" << file_id << ", begin pos:" << i << \
				", len:" << try_to_read_len << ", ok");
	}
	ret = notify_mds_file_info(dir_id, file_id, 0, NOTIFY_MDS_FILE_INFO_REASON_REPLICATE);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", notify_mds_file_info error, dir_id:" << dir_id << ", file_id:" << \
				file_id << ", ret:" << ret);
	}
	return 1; 
}

int DataSvr::notify_mds_file_info(int32_t dir_id, int64_t file_id, int delete_flag, int notify_reason) {
	if (dir_id < 0 || file_id < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", invalid args");
		return -1;
	}
	string content_file_path = get_file_hash_path(_root_data_path.c_str(), dir_id, file_id);
	string control_file_path = get_control_file_hash_path(_root_data_path.c_str(), dir_id, file_id);
	MetadataSvrClient * mdsc = _mdscg.get_meta_data_svr_client(_my_mds_id);
	if (NULL == mdsc) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not get meta data svr client, dir_id:" << dir_id << \
				", file_id:" << file_id << ", delete flag:" << delete_flag << ", notify_reason:" << \
				notify_reason);
		return -2;
	}
	vector<uint32_t> crcs;
	int64_t file_size = -1;
	if (1 != delete_flag) {
		struct stat s;
		int ret = stat(content_file_path.c_str(), &s);
		if (0 != ret) {
			LOG4CXX_ERROR(_logger, __FUNCTION__ << ", stat error, path:" << content_file_path << ", errno:" << \
					errno << ", dir_id:" << dir_id << ", file_id:" << file_id << ", notify_reason:" << \
					notify_reason);
			return -3;
		}
		file_size = s.st_size;
		ret = parse_control_file(control_file_path.c_str(), crcs);
		if (ret < 0) {
			LOG4CXX_ERROR(_logger, __FUNCTION__ << ", parse control file error, file:" << control_file_path << \
					", ret:" << ret << ", dir_id:" << dir_id << ", file_id:" << file_id << \
					", notify_reason:" << notify_reason);
			return -4;
		}
	}
	int ret = mdsc->data_svr_notify_data_file_info(dir_id, file_id, _my_ds_id, file_size, crcs, delete_flag, notify_reason);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", data_svr_notify_data_file_info error, dir_id:" << dir_id << \
		", file_id:" << file_id << ", notify_reason:" << notify_reason << ", ret:" << ret);
		return -5;
	}
	LOG4CXX_INFO(_logger, __FUNCTION__ << ", ok, dir_id:" << dir_id << ", file_id:" << file_id << ", file_size:" << \
			file_size << ", delete_flag:" << delete_flag << ", notify_reason:" << notify_reason);
	return 1;

}

int DataSvr::handle_truncate(json_object * j, string & response, string & extra_response, const int fd, bool & is_network_error) {
        json_object * j_dir_id = json_object_object_get(j, "DIR_ID");
        json_object * j_file_id = json_object_object_get(j, "FILE_ID");
        json_object * j_size = json_object_object_get(j, "SIZE");
        if( NULL == j_dir_id || NULL == j_file_id || NULL == j_size ) {
                LOG4CXX_ERROR(_logger, __FUNCTION__ << ", args missing:" << json_object_to_json_string(j));
                response = build_error_response(DS_ERR_INVALID_ARGS, "");
                return -1;
        }
	int dir_id = json_object_get_int(j_dir_id);
	int64_t file_id = json_object_get_int(j_file_id);
	int64_t size = json_object_get_int(j_size);
	if (size < 0) {
		response = build_ok_response();
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", size is negative, request :" << json_object_to_json_string(j));
		return -2;
	}
	string content_file_path = get_file_hash_path(_root_data_path.c_str(), dir_id, file_id);
	string control_file_path = get_control_file_hash_path(_root_data_path.c_str(), dir_id, file_id);
	struct stat s;
	int ret = stat(content_file_path.c_str(), &s);
	if (0 != ret) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", stat file error, file:" << content_file_path << ", errno:" << errno << \
				", request:" << json_object_to_json_string(j));
                response = build_error_response(DS_ERR_OS_FILE_OP_ERR);
		return -3;
	}
	if (s.st_size == size) {
		LOG4CXX_INFO(_logger, __FUNCTION__ << ", size is the same, no need to truncate, request:" << \
                                json_object_to_json_string(j));
		response = build_ok_response();
		return 1;
	}
	ret = truncate(content_file_path.c_str(), size); 
	if (0 != ret) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", truncate file error, file:" << content_file_path << ",  errno:" << \
				errno << ", request:" << json_object_to_json_string(j));
                response = build_error_response(DS_ERR_OS_FILE_OP_ERR);
		return -4;
	}
	ret = total_rebuild_crc(dir_id, file_id);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", total_rebuild_crc error, ret:" << ret << ", request:" << \
				json_object_to_json_string(j));
                response = build_error_response(DS_ERR_OS_FILE_OP_ERR);
		return -5;
	}
	ret = notify_mds_file_info(dir_id, file_id, 0, NOTIFY_MDS_FILE_INFO_REASON_FILE_OP);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", notify_mds_file_info error, ret:" << ret << ", request:" << \
				json_object_to_json_string(j));
		response = build_error_response(DS_ERR_NOTIFY_MDS_ERROR);
		return -6;
	} 
	response = build_ok_response();
	LOG4CXX_INFO(_logger, __FUNCTION__ << ", ok, request:" << json_object_to_json_string(j));
	return 2;
}
